package com.xkt.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;

import java.util.Optional;

/**
 * Description:
 *
 * @Create 2023/12/14 23:42
 * @Version 1.0
 */
@Slf4j
public class QueueConsumer {

    @KafkaListener(topics = "test", groupId = "consumer")
    public void doConsumer(ConsumerRecord record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {

        Optional message = Optional.ofNullable(record.value());

        if (message.isPresent()) {
            try {
                Object msg = message.get();
                log.info("[kafka-消费] doConsumer 消费了： Topic:" + topic + ",Message:" + msg);
                // queueDataProcess.doExec(JSON.parseObject(msg.toString(), Order.class))
                boolean res = true;
                if (res) {
                    ack.acknowledge();
                }
            } catch (Exception ex) {
                log.error("[kafka-消费异常] doConsumer Error {} ", ex);
            }

        }

    }

}
